package org.huangrui.spark.scala.core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Top10 热门品类中每个品类的 Top10 活跃 Session 统计
 *
 * @Author hr
 * @Create 2024-10-20 6:04 
 */
object HotCategoryTop10SessionAnalysis {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("spark")
    val sc = new SparkContext(conf)
    val actionRDD = sc.textFile("data/user_visit_action.txt")
    actionRDD.cache()
    val top10Ids: Array[String] = top10Category(actionRDD)
    // 1. 过滤原始数据,保留点击和前10品类ID
    val filterRdd = actionRDD.filter {
      action: String => {
        val ds = action.split("_")
        if (ds(6) != "-1") {
          top10Ids.contains(ds(6))
        } else {
          false
        }
      }
    }
    // 2. 根据品类ID和sessionid进行点击量的统计
    val reduceRdd = filterRdd.map {
      action: String => {
        val ds = action.split("_")
        ((ds(6), ds(2)), 1)
      }
    }.reduceByKey(_ + _)

    // 3. 将统计的结果进行结构的转换
    //  ((品类ID，sessionId),sum) => (品类ID,(sessionId, sum))
    val mapRdd = reduceRdd.map {
      case ((id, session), sum) => (id, (session, sum))
    }

    // 4. 相同的品类进行分组
    val groupRdd = mapRdd.groupByKey()
    // 5. 将分组后的数据进行点击量的排序，取前10名
    val sortRdd = groupRdd.mapValues {
      iter: Iterable[(String, Int)] => {
        iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(10)
      }
    }

    println(sortRdd.collect().mkString("\n"))

    sc.stop()
  }

  def top10Category(actionRDD: RDD[String]): Array[String] = {
    val flatRdd: RDD[(String, (Int, Int, Int))] = actionRDD.flatMap {
      action: String => {
        val datas = action.split("_")
        if (datas(6) != "-1") {
          // 点击的场合
          List((datas(6), (1, 0, 0)))
        } else if (datas(8) != "null") {
          // 下单的场合
          val ids = datas(8).split(",")
          ids.map(id => (id, (0, 1, 0)))
        } else if (datas(10) != "null") {
          // / 支付的场合
          val ids = datas(10).split(",")
          ids.map(id => (id, (0, 0, 1)))
        } else {
          Nil
        }
      }
    }
    val reduceRdd = flatRdd.reduceByKey((a, b) => (a._1 + b._1, a._2 + b._2, a._3 + b._3))
    reduceRdd.sortBy(x => x._2, false).map(_._1).take(10)
  }
}
